package ari.com.messaging.queue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class QueueImplementationTest {
	static BlockingQueue<String> queue = new LinkedBlockingQueue<String>();

	public static void main(final String[] args) throws InterruptedException {
		final String[] messages = new String[] { "ABC", "DEF", "GHI", "JKL",
				"MNO" };
		System.out.println("[MAIN] starting the Main thread");

		new Thread(new Consumer<String>(queue, "AMITH")).start();
		System.out.println("[MAIN] starting the Consumer");
		new Thread(new Consumer<String>(queue, "SIMITH")).start();
		System.out.println("[MAIN] starting the Consumer");
		new Thread(new Consumer<String>(queue, "SANIL")).start();
		System.out.println("[MAIN] starting the Consumer");
		new Thread(new Producer<String>(queue, messages)).start();
		System.out.println("[MAIN] starting the Producer");
		while (true) {
			;
		}

	}

	static class Producer<E> implements Runnable {
		BlockingQueue<E> queue;
		E[] arrayOfE;

		Producer(final BlockingQueue<E> queue, final E[] arrayOfE) {
			this.queue = queue;
			this.arrayOfE = arrayOfE;
		}

		public void run() {
			for (final E e : arrayOfE) {
				System.out.println("[Producer] adding an element " + e);
				queue.add(e);
				System.out.println("element added :" + e);
				try {
					System.out.println("[Producer] going to sleep");
					Thread.sleep(3000);
					System.out.println("[Producer] is back from sleep");
				} catch (final InterruptedException e1) {
					e1.printStackTrace();
				}
			}
		}
	}

	static class Consumer<E> implements Runnable {
		BlockingQueue<E> queue;
		String name;

		Consumer(final BlockingQueue<E> queue, final String name) {
			this.queue = queue;
			this.name = name;
		}

		public void run() {
			E e = null;
			try {
				while (true) {
					System.out.println("[CONSUMER-" + name
							+ "] before calling take()");
					e = queue.take();
					System.out.println("[CONSUMER-" + name
							+ "] element retreived :" + e);
				}
			} catch (final InterruptedException e1) {
				e1.printStackTrace();
			}
		}
	}
}
